package com.arch.flink.archetype;

import org.apache.flink.table.functions.AggregateFunction;


public class GetStatusLatest extends AggregateFunction<String, TimeVersionAndStringStatus> {

    @Override
    public String getValue(TimeVersionAndStringStatus accumulator) {
        return accumulator.getStatus();
    }


//    @FunctionHint(
//            accumulator = @DataTypeHint("ROW<status STRING,timeVersion BIGINT>"),
//            output = @DataTypeHint("BIGINT")
//    )
    public void accumulate(TimeVersionAndStringStatus acc, String status, long versionTime) {
        if (versionTime >= acc.getTimeVersion()) {
            acc.setStatus(status);
            acc.setTimeVersion(versionTime);
        }
    }

    @Override
    public TimeVersionAndStringStatus createAccumulator() {
        TimeVersionAndStringStatus acc  = new TimeVersionAndStringStatus();
        acc.setTimeVersion(Long.MIN_VALUE);
        return acc;
    }
}
